package cn.lgwen.candy.runtime.driver;

import cn.lgwen.candy.runtime.operator.BufferingCollector;
import cn.lgwen.candy.runtime.operator.Collector;
import cn.lgwen.candy.runtime.operator.Fetcher;
import cn.lgwen.candy.runtime.operator.SourceOperator;
import cn.lgwen.candy.runtime.queue.Queue;

import java.util.LinkedList;
import java.util.List;

/**
 * 2019/12/24
 * aven.wu
 * danxieai258@163.com
 */
public class SourceDriver<O> implements Driver {

    // source 结束后应该向jobmanager 发个通知等待所有operator 都执行完毕之后 退出程序
    private volatile boolean running = true;

    private List<Queue<O>> forwardQueues;

    // 可能是一个不会返回的方法
    private SourceOperator<O> operator;

    private boolean checkpoint;

    private int parallel = 1;
    public SourceDriver(SourceOperator<O> operator) {
        this.operator = operator;
        forwardQueues = new LinkedList<>();
    }

    public void registerForward(Queue<O> queue) {
        forwardQueues.add(queue);
    }

    @Override
    public void run() {
        operator.checkpoint(checkpoint());
        Fetcher<O> fetcher = new Fetcher<>();
        DataSourceTask<O> task = new DataSourceTask<>(fetcher, operator);
        Thread sourceTask = new Thread(task);
        sourceTask.start();
        while (running && !forwardQueues.isEmpty()) {
            O record = fetcher.pollNext();
            forwardQueues.forEach(queue -> queue.add(record));
        }
    }

    @Override
    public void parallel(int parallel) {
        this.parallel = parallel;
    }

    @Override
    public int parallel() {
        return 1;
    }

    @Override
    public void checkpoint(boolean checkpoint) {
        this.checkpoint = checkpoint;
    }

    @Override
    public boolean checkpoint() {
        return checkpoint;
    }

    public void cancel() {
        this.running = false;
    }
}
